package main;

import helper.CompositeKey;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class ValueCombiner extends Reducer<CompositeKey, IntWritable, CompositeKey, IntWritable> {
    
    private CompositeKey outKey = new CompositeKey();
    
    public void reduce(CompositeKey key, Iterable<IntWritable> values, Context context) 
            throws IOException, InterruptedException {

//        int sum = 0;
//        for (IntWritable val : values) {
//            sum += val.get();
//        }
//        result.set(sum);
//        context.write(key, result);
        Map<Integer, Integer> similarMap = new HashMap<Integer, Integer>();
        
        for (IntWritable val : values) {
            int similar = key.getSimilar();
            if(similarMap.containsKey(similar)) {
                similarMap.put(similar, similarMap.get(similar)+val.get());
            } else {
                similarMap.put(similar, val.get());
            }
//            System.out.println(key.toString() + " === " + similarMap.toString());
        }
        
        outKey.setUser(key.getUser());
        for (Entry<Integer, Integer> entry : similarMap.entrySet()) {
            outKey.setSimilar(entry.getKey());
            context.write(outKey, new IntWritable(entry.getValue()));
        }
    } 
    
}
